#include "storage/buf_ext.h"
#include "storage/fd.h"

#include "common/hashfn.h"
#include "utils/wait_event.h"


/*
 * KPHCPciKeyCmp
 *
 * compute pci key 
 */
static 
inline uint32 KPHCPciKeyCmp(KPHCCfsBufferKey *keya, KPHCCfsBufferKey *keyb)
{
    return (uint32)gPciBufCtx->hashtbl.match((void *)keya, (void *)keyb, sizeof(KPHCCfsBufferKey));
}

/*
 * KPHCPciLockCount
 *
 * Getting Lock Count 
 */
/*static
uint32 KPHCPciLockCount()
{
    Size buffer_size = 0;
    uint32 ctrlCount = 0;
    buffer_size = KPHCPciBufferSize();
    ctrlCount = (uint32)((buffer_size - sizeof(KPHCPciPageBuffCtx)) /
        (sizeof(KPHCPciPageCtrl) + BLCKSZ + 3 * sizeof(KPHCPciHashBucket)));
    return (PCI_LRU_LIST_NUM + ctrlCount + ctrlCount * 3);
}*/

/*
 * KPHCPciHashCode
 *
 * Getting Hash Code 
 */
static 
inline uint32 KPHCPciHashCode(KPHCCfsBufferKey *key)
{
    return gPciBufCtx->hashtbl.hash((void *)key, sizeof(KPHCCfsBufferKey));
}

/*
 * KPHCPciBufFindFromBucket
 *
 * Getting Ctrl from bucket
 */
static
KPHCPciPageCtrl *KPHCPciBufFindFromBucket(KPHCPciHashBucket *bucket, KPHCCfsBufferKey *key)
{
    KPHCPciPageCtrl *ctrl = NULL;

    uint32 ctrlId = bucket->firstCtrlId;
    while (ctrlId != PCI_INVALID_ID) {
        ctrl = PCI_GET_CTRL_BY_ID(gPciBufCtx, ctrlId);
        if (KPHCPciKeyCmp(key, &ctrl->pciKey) == 0) {
            return ctrl;
        }
        ctrlId = ctrl->bckNext;
    }

    return NULL;
}

/*
 * KPHCPciLruRemove
 *
 * remove Ctrl page from bucket
 */
static
void KPHCPciLruRemove(KPHCPciLruList *lru, KPHCPciPageCtrl *item)
{
    KPHCPciPageCtrl *prev = NULL;
    KPHCPciPageCtrl *next = NULL;

    if (item->lruPrev != PCI_INVALID_ID) {
        prev = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->lruPrev);
        prev->lruNext = item->lruNext;
    } else {
        lru->first = item->lruNext;
    }

    if (item->lruNext != PCI_INVALID_ID) {
        next = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->lruNext);
        next->lruPrev = item->lruPrev;
    } else {
        lru->last = item->lruPrev;
    }

    item->lruPrev = PCI_INVALID_ID;
    item->lruNext = PCI_INVALID_ID;
    item->state = CTRL_STATE_INIT;
    lru->count--;

    return ;
}

/*
 * KPHCPciBufRemoveFromBucket
 *
 * remove ctrl page from bucket
 */
static
void KPHCPciBufRemoveFromBucket(KPHCPciHashBucket *bucket, KPHCPciPageCtrl *item)
{
    KPHCPciPageCtrl *prev = NULL;
    KPHCPciPageCtrl *next = NULL;

    if (item->bckPrev != PCI_INVALID_ID) {
        prev = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->bckPrev);
        prev->bckNext = item->bckNext;
    } else {
        bucket->firstCtrlId = item->bckNext;
    }

    if (item->bckNext != PCI_INVALID_ID) {
        next = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->bckNext);
        next->bckPrev = item->bckPrev;
    }

    item->bckPrev = PCI_INVALID_ID;
    item->bckNext = PCI_INVALID_ID;
    item->bckId = PCI_INVALID_ID;

    memset(&item->pciKey, 0x00, sizeof(item->pciKey));

    bucket->ctrlCount--;
}

/*
 * KPHCPciBufFindFromBucket
 *
 * remove buf from bucket
 */
static
bool KPHCPciBufRecycleCoreTry(KPHCPciPageCtrl *item)
{
    KPHCPciHashBucket *bucket = NULL;
    if ((pg_atomic_read_u32(&item->refNum) > 0) || (pg_atomic_read_u32(&item->touchNum) >= 3)) {
        return false;
    }

    bucket = PCI_GET_BUCKET_BY_ID(gPciBufCtx, item->bckId);
    if (!LWLockConditionalAcquire(&bucket->lock, LW_EXCLUSIVE)) {
        return false; 
    }

    if ((pg_atomic_read_u32(&item->refNum) > 0) || (pg_atomic_read_u32(&item->touchNum) >= 3)) {
        LWLockRelease(&bucket->lock);
        return false;
    }

    KPHCPciBufRemoveFromBucket(bucket, item);
    pg_atomic_write_u32(&item->touchNum, 0);
    LWLockRelease(&bucket->lock);
    return true;
}


/*
 * KPHCPciLruPop
 *
 * get ctrl page from LRU
 */
static
KPHCPciPageCtrl *KPHCPciLruPop(KPHCPciLruList *lru)
{
    KPHCPciPageCtrl *item = NULL;

    if (lru->count == 0) {
        return NULL;
    }
    (void)LWLockAcquire(&lru->lock, LW_EXCLUSIVE);

    if (lru->count == 0) {
        LWLockRelease(&lru->lock);
        return NULL;
    }

    item = PCI_GET_CTRL_BY_ID(gPciBufCtx, lru->last);
    KPHCPciLruRemove(lru, item);
    LWLockRelease(&lru->lock);
    return item;
}

/*
 * KPHCPciLruPop
 *
 * remove ctrl page from LRU
 */
static
KPHCPciPageCtrl *KPHCPciBufRecycleCore(KPHCPciLruList *mainLru)
{
    uint32 step = 0;
    KPHCPciPageCtrl *item = NULL;
    KPHCPciPageCtrl *res = NULL;
    uint32 expected_val = 0;
    uint32 currCtrlId = 0;

    if (!LWLockConditionalAcquire(&mainLru->lock, LW_EXCLUSIVE)) {
        return res;
    }

    currCtrlId = mainLru->last;
    while (currCtrlId != PCI_INVALID_ID) {
        if (step >= 1024) {
            break;
        }

        item = PCI_GET_CTRL_BY_ID(gPciBufCtx, currCtrlId);
        if (KPHCPciBufRecycleCoreTry(item)) {
            KPHCPciLruRemove(mainLru, item);
            res = item;
            break;
        }

        expected_val = pg_atomic_read_u32(&item->touchNum);
        pg_atomic_compare_exchange_u32_impl(&item->touchNum, &expected_val, expected_val/2);

        step++;
        currCtrlId = item->lruPrev;
    }

    LWLockRelease(&mainLru->lock);
    return res;
}

/*
 * KPHCPciBufRecycle
 *
 * get free page
 */
static
KPHCPciPageCtrl *KPHCPciBufRecycle(uint32 randVal)
{
    KPHCPciPageCtrl *item = NULL;
    uint8 randStart = randVal % PCI_PART_LIST_NUM;

    while (1) {
        for (uint8 i = 0; i < PCI_PART_LIST_NUM; i++) {
            item = KPHCPciLruPop(&gPciBufCtx->freeLru[(randStart + i) % PCI_PART_LIST_NUM]);
            if (item != NULL) {
                return item;
            }
        }
        for (uint8 i = 0; i < PCI_PART_LIST_NUM; i++) {
            item = KPHCPciBufRecycleCore(&gPciBufCtx->mainLru[(randStart + i) % PCI_PART_LIST_NUM]);
            if (item != NULL) {
                return item;
            }
            if (gPciBufCtx->freeLru[i].count != 0) {
                break;
            }
        }
    }
}



/*
 * KPHCPciBufLoadPage
 *
 * Loading pci page from file 
 */
static
void KPHCPciBufLoadPage(KPHCPciPageCtrl *item, const KPHCExtentLocation location, KPHCCfsBufferKey *pci_key)
{
    int nbytes = 0;
    memcpy(&item->pciKey, (char *)pci_key, sizeof(KPHCCfsBufferKey));
    nbytes = FileRead(location.fd, (char *)item->pciPage, BLCKSZ, 
                           location.headerNum * BLCKSZ, (uint32)WAIT_EVENT_DATA_FILE_READ);
    if (nbytes != BLCKSZ) {
        item->loadStatus = CTRL_PAGE_LOADED_ERROR;
        ereport(DEBUG5, (errcode(ERRCODE_DATA_CORRUPTED),
            errmsg("Failed to pci_buf_load_page %s, headerNum: %u.", FilePathName(location.fd), location.headerNum)));
        return;
    }
    item->loadStatus = CTRL_PAGE_IS_LOADED;
    return;
}

/*
 * KPHCPciBufAddIntoBucket
 *
 * adding pci ctrl page into bucket 
 */
static
void KPHCPciBufAddIntoBucket(KPHCPciHashBucket *bucket, KPHCPciPageCtrl *item)
{
    KPHCPciPageCtrl *head = NULL;
    if (bucket->ctrlCount == 0) {
        item->bckId = bucket->bckId;
        item->bckPrev = PCI_INVALID_ID;
        item->bckNext = PCI_INVALID_ID;
        bucket->firstCtrlId = item->ctrlId;
        bucket->ctrlCount++;
        return;
    }
    head = PCI_GET_CTRL_BY_ID(gPciBufCtx, bucket->firstCtrlId);
    item->bckId = bucket->bckId;
    item->bckPrev = PCI_INVALID_ID;
    item->bckNext = head->ctrlId;
    head->bckPrev = item->ctrlId;
    bucket->firstCtrlId = item->ctrlId;
    bucket->ctrlCount++;
}

/*
 * KPHCPciLruPush
 *
 * adding pci ctrl page into LRU 
 */
static
void KPHCPciLruPush(KPHCPciLruList *lru, KPHCPciPageCtrl *item, KPHCCtrlState state)
{
    (void)LWLockAcquire(&lru->lock, LW_EXCLUSIVE);
    KPHCPciLruPushNoLock(lru, item, state);
    LWLockRelease(&lru->lock);
}

/*
 * KPHCPciBufReadPage
 *
 * Reading page from Pci buffer
 */
KPHCPciPageCtrl *KPHCPciBufReadPage(const KPHCExtentLocation location, LWLockMode lockMode, KPHCPciBufferReadMode readMode)
{
    KPHCCfsBufferKey key = {{location.relFileNode.spcNode, location.relFileNode.dbNode, location.relFileNode.relNode,
                         location.relFileNode.bucketNode},
        location.extentNumber};
    uint32 hashcode = 0;
    KPHCPciHashBucket *bucket = NULL;
    KPHCPciPageCtrl *ctrl = NULL;
    KPHCPciPageCtrl *item = NULL;

    hashcode = KPHCPciHashCode(&key);
    bucket = PCI_GET_BUCKET_BY_HASH(gPciBufCtx, hashcode);
    (void)LWLockAcquire(&bucket->lock, LW_EXCLUSIVE);
    ctrl = KPHCPciBufFindFromBucket(bucket, &key);
    if (ctrl != NULL) {
        (void)LWLockAcquire(&ctrl->contentLock, lockMode);
        if (ctrl->loadStatus != CTRL_PAGE_IS_LOADED && readMode == PCI_BUF_NORMAL_READ) {
            KPHCPciBufLoadPage(ctrl, location, &key);
        }
        (void)pg_atomic_fetch_add_u32(&ctrl->refNum, 1);
        (void)pg_atomic_fetch_add_u32(&ctrl->touchNum, 3);
        LWLockRelease(&bucket->lock);
        return ctrl;
    }
    LWLockRelease(&bucket->lock);
    item = KPHCPciBufRecycle(location.extentNumber);
    (void)LWLockAcquire(&bucket->lock, LW_EXCLUSIVE);
    ctrl = KPHCPciBufFindFromBucket(bucket, &key);
    if (ctrl != NULL) {
        (void)LWLockAcquire(&ctrl->contentLock, lockMode);
        if (ctrl->loadStatus != CTRL_PAGE_IS_LOADED && readMode == PCI_BUF_NORMAL_READ) {
            KPHCPciBufLoadPage(ctrl, location, &key);
        }
        (void)pg_atomic_fetch_add_u32(&ctrl->refNum, 1);
        (void)pg_atomic_fetch_add_u32(&ctrl->touchNum, 1);
        LWLockRelease(&bucket->lock);
        KPHCPciLruPush(&gPciBufCtx->freeLru[item->ctrlId % PCI_PART_LIST_NUM], item, CTRL_STATE_FREE);
        return ctrl;
    }
    pg_atomic_write_u32(&item->refNum, 1);
    pg_atomic_write_u32(&item->touchNum, 3);
    memset((void*)item->pciPage, 0, BLCKSZ);
    item->loadStatus = CTRL_PAGE_NO_LOAD;
    if (readMode == PCI_BUF_NORMAL_READ) {
        KPHCPciBufLoadPage(item, location, &key);
    }
    KPHCPciBufAddIntoBucket(bucket, item);
    (void)LWLockAcquire(&item->contentLock, lockMode);
    LWLockRelease(&bucket->lock);
    KPHCPciLruPush(&gPciBufCtx->mainLru[item->ctrlId % PCI_PART_LIST_NUM], item, CTRL_STATE_USED);
    return item;
}


/*
 * KPHCPciBufFreePage
 *
 * Free page 
 */
void KPHCPciBufFreePage(KPHCPciPageCtrl *ctrl, const KPHCExtentLocation location, bool need_write)
{
    int nbytes = 0;
    if (need_write) {
        nbytes = FileWrite(location.fd, (char *)ctrl->pciPage, BLCKSZ, location.headerNum * BLCKSZ,
                                (uint32)WAIT_EVENT_DATA_FILE_WRITE);
        if (nbytes != BLCKSZ) {
            (void)pg_atomic_fetch_sub_u32(&ctrl->refNum, 1);
            ctrl->loadStatus = CTRL_PAGE_LOADED_ERROR;
            LWLockRelease(&ctrl->contentLock);

            ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                errmsg("Failed to KPHCPciBufFreePage %s", FilePathName(location.fd))));
            return;
        }
        ctrl->loadStatus = CTRL_PAGE_IS_LOADED;
    }

    (void)pg_atomic_fetch_sub_u32(&ctrl->refNum, 1);
    LWLockRelease(&ctrl->contentLock);
}